package com.hujiang.basic.search.util;

import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.hujiang.basic.search.common.component.HJElasticSearchClient;
import com.hujiang.basic.search.common.converter.req.BulkIndexRequestConverter;
import com.hujiang.basic.search.common.exception.EsSettingException;
import com.hujiang.basic.search.common.exception.FullIndexException;
import com.hujiang.basic.search.common.exception.ParameterException;
import com.hujiang.basic.search.model.HJBulkIndexRequest;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesResponse;
import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.cluster.metadata.AliasMetaData;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.indices.IndexAlreadyExistsException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Repository;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

@Slf4j
@Repository
public class IndexEsDaoImpl implements IndexEsDao {

    @Autowired
    private HJElasticSearchClient esClient;

    @Autowired
    private BulkIndexRequestConverter bulkConverter;

    @Value("${fullIndex.indexRefreshInterval:1s}")
    private String indexRefreshInterval;


    @Override
    public String getIndexName(String indexAlias) {

        GetAliasesRequest req = new GetAliasesRequest();
        GetAliasesResponse getAliasesResponse = esClient.execute(c -> c.admin().indices().getAliases(req)).actionGet();
        ImmutableOpenMap<String, List<AliasMetaData>> indexAliasesMaps = getAliasesResponse.getAliases();
        List<String> indexs = Lists.newArrayList();
        indexAliasesMaps.forEach(e -> {
            e.value.stream().filter(meta -> meta.alias().equals(indexAlias)).forEach(c -> indexs.add(e.key));
        });
        if (indexs.size() != 1) {
            throw new ParameterException("Index 与 Alias 不是一一对应,无法进行切换,请检查。Alias: " + indexAlias +
                    ", Index: " + indexs);
        }
        return indexs.get(0);
    }

    /**
     * (non-Javadoc)
     *
     * @see com.hujiang.basic.search.util.IndexEsDao#
     * lang.String)
     */
    @Override
    public Map<String, String> getIndexMappings(String indexName) {
        GetMappingsRequest req = new GetMappingsRequest().indices(indexName);
        GetMappingsResponse resp = esClient.execute(c -> c.admin().indices().getMappings(req)).actionGet();
        ImmutableOpenMap<String, ImmutableOpenMap<String, MappingMetaData>> mappings = resp.getMappings();
        ImmutableOpenMap<String, MappingMetaData> indexMeta = mappings.get(indexName);
        if (indexMeta == null) {
            throw new EsSettingException("Failed to get index mappings with index name(" + indexName + ")");
        }
        Map<String, String> indexMappings = Maps.newHashMap();
        for (ObjectObjectCursor<String, MappingMetaData> cursor : indexMeta) {
            indexMappings.put(cursor.key, cursor.value.source().toString());
        }
        return indexMappings;
    }

    @Override
    public Map<String, String> getIndexSettings(String indexName) {
        GetSettingsRequest req = new GetSettingsRequest().indices(indexName);
        GetSettingsResponse resp = esClient.execute(c -> c.admin().indices().getSettings(req)).actionGet();
        ImmutableOpenMap<String, Settings> settings = resp.getIndexToSettings();
        Settings setting = settings.get(indexName);
        if (setting == null) {
            throw new EsSettingException("Failed to get index settings with index name(" + indexName + ")");
        }
        return new HashMap<>(setting.getAsMap());
    }

    /**
     * (non-Javadoc)
     *
     * @see com.hujiang.basic.search.util.IndexEsDao#
     * String, java.util.Map)
     */
    @Override
    public void createIndex(String indexName, Map<String, String> indexMappings, Map<String, String> indexSettings) {
        indexSettings.put("index.refresh_interval", "-1");

        CreateIndexRequest req = new CreateIndexRequest().index(indexName).settings(indexSettings);
        for (Entry<String, String> entry : indexMappings.entrySet()) {
            req.mapping(entry.getKey(), entry.getValue());
        }
        try {
            esClient.execute(c -> c.admin().indices().create(req)).actionGet();
        } catch (IndexAlreadyExistsException e) {
            log.info("The index{} is already exists, so ignore it.", indexName);
        }
    }

    @Override
    public void createIndex(String indexName, String indexConfig, String indexAlias) {
        CreateIndexRequest req = new CreateIndexRequest().index(indexName).source(indexConfig);
        try {
            esClient.execute(c -> c.admin().indices().create(req)).actionGet();
            this.addAlias(indexName, indexAlias);
        } catch (IndexAlreadyExistsException e) {
            log.info("The index{} is already exists, so ignore it.error:{}", indexName,e);
        }
    }

    /**
     * (non-Javadoc)
     *
     * @see com.hujiang.basic.search.util.IndexEsDao#
     */
    @Override
    public void addBulkIndexData(String newIndexName, HJBulkIndexRequest hjBulkIndexRequest) {
        long start = System.currentTimeMillis();
        log.info("Start to bulk add index({}) data({}).", newIndexName, hjBulkIndexRequest.getSources().size());
        hjBulkIndexRequest.setIndex(newIndexName);
        BulkResponse bulkResponse = esClient.bulk(bulkConverter.convert(hjBulkIndexRequest));
        log.info("End({}) to bulk add index({}) data({}).", new Date().getTime() - start, newIndexName,
                hjBulkIndexRequest.getSources().size());
        if (bulkResponse.hasFailures()) {
            String errorMessage = bulkResponse.getItems()[0].getFailureMessage() + ": " + bulkResponse
                    .buildFailureMessage();
            log.error(errorMessage);
            throw new FullIndexException(errorMessage);
        }
    }

    /**
     * (non-Javadoc)
     *
     * @see com.hujiang.basic.search.util.IndexEsDao#
     * lang.String, java.lang.String, java.lang.String)
     */
    @Override
    public void changeIndexAlias(String indexName, String newIndexName, String indexAlias) {
        // update index.refresh_interval
        Map<String, String> settings = new HashMap<>();
        settings.put("index.refresh_interval", indexRefreshInterval);
        UpdateSettingsRequest req = new UpdateSettingsRequest().indices(newIndexName).settings(settings);
        esClient.execute(c -> c.admin().indices().updateSettings(req)).actionGet();

        // change alias
        IndicesAliasesRequest switchAliasReq = new IndicesAliasesRequest();
        switchAliasReq.addAlias(indexAlias, newIndexName);
        switchAliasReq.removeAlias(indexName, indexAlias);
        esClient.execute(c -> c.admin().indices().aliases(switchAliasReq)).actionGet();

        // close old index
        closeIndex(indexName);

    }


    @Override
    public void closeIndex(String indexName) {
        CloseIndexRequest closeReq = new CloseIndexRequest().indices(indexName);
        esClient.execute(c -> c.admin().indices().close(closeReq)).actionGet();
    }

    @Override
    public long getCount(String index) {
        SearchResponse response = esClient.getCount(index);
        return response.getHits().totalHits();
    }

    @Override
    public void flushAndRefresh(String index) {
        esClient.flushAndRefresh(index);
    }

    @Override
    public void addAlias(String indexName, String indexAlias) {
        IndicesAliasesRequest switAliasReq = new IndicesAliasesRequest();
        switAliasReq.addAlias(indexAlias, indexName);
        esClient.execute(c -> c.admin().indices().aliases(switAliasReq)).actionGet();
    }
}
